package com.example.mq.server.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.server.VirtualHost;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

public class ConsumerManager {
    // 使用虚拟主机对象操作数据
    private VirtualHost virtualHost;
    // 使用一个阻塞队列来存储队列的名字，线程池可以直接从这里取
    private BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();
    // 扫描线程，用于扫描哪些队列的消息需要消费
    private Thread scannerThread = null;
    // 创建线程池，用于执行回调函数
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);

    public ConsumerManager(VirtualHost virtualHost) {
        this.virtualHost = virtualHost;
        // 开启一个扫描线程
        scannerThread = new Thread(() -> {
            while (true) {
                // 一直扫描，只有程序结束才停止
                // 拿令牌，从对应的队列找消息
                try {
                    String queueName = tokenQueue.take();
                    // 查询队列是否存在
                    MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 取令牌时，发现队列不存在！queueName = " + queueName);
                    }
                    synchronized (queue) {
                        // 将消息消费
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
        // 设置为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    // 通知订阅者消息已经来了
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 判断队列是否存在
        MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不成功！queueName = " + queueName);
        }
        // 将消费者添加进队列
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            // 如果在消费者还没有来之前队列中已经有消息了，就需要先消费掉之前的消息
            int queueMessageNum = virtualHost.getMemoryDataCenter().getQueueMessageNum(queueName);
            for (int i = 0; i < queueMessageNum; i++) {
                // 有多少就消费多少
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {
        // 1.按照轮询的方式取出消费者
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null) {
            // 没有消费者就不推送消息
            return;
        }
        // 2.从队列中取出消息
        Message message = virtualHost.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            // 没有消息不推送
            return;
        }
        // 3.将消息带到消费者的回调函数中，丢到线程池中
        workerPool.submit(() -> {
            try {
                // 1.将消息放入待确认的集合中，防止因为出错造成消息丢失
                virtualHost.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                // 2.执行回调
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(),
                       message.getBasicProperties(), message.getBody());
                // 3.判断autoAck
                //   3.1 如果是自动ack，不抛异常直接删除数据
                if (luckyDog.isAutoAck()) {
                    // 1) 删除硬盘消息
                    if (message.getDeliverMode() == 2) {
                        virtualHost.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 2) 删除内存消息
                    virtualHost.getMemoryDataCenter().deleteMessage(message.getMessageId());
                    // 3) 删除待确认集合消息
                    virtualHost.getMemoryDataCenter().deleteMessageWaitAck(queue.getName(), message.getMessageId());
                }
                //   3.2 如果不是自动ack，消费者手动调用basicAck
                //virtualHost.basicAck();
                System.out.println("[ConsumerManager] 消息成功消费！queueName = " + queue.getName());
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}
